package com.baichen.trade.rocketmq;

import com.baichen.trade.constant.MQEnums;
import com.baichen.trade.exception.TradeMQException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
/**
 * @Date: 2019-11-07 23:22
 * @Author: baichen
 * @Description 生产者
 */
public class TradeMQProducer {
    public static final Logger LOGGER = LoggerFactory.getLogger(TradeMQProducer.class);
    private DefaultMQProducer producer;
    private String groupName;
    private String namesrvAddr;
    private int maxMessageSize = 1024 * 1024 * 4;
    private int sendMsgTimeout = 10000;
    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public void setMaxMessageSize(int maxMessageSize) {
        this.maxMessageSize = maxMessageSize;
    }

    public void setSendMsgTimeout(int sendMsgTimeout) {
        this.sendMsgTimeout = sendMsgTimeout;
    }

    public void init() throws TradeMQException{
        if(StringUtils.isBlank(this.groupName)){
            throw new TradeMQException("groupName is blank!");
        }
        if(StringUtils.isBlank(this.namesrvAddr)){
            throw new TradeMQException("namesrvAddr is blank!");
        }
        this.producer=new DefaultMQProducer(this.groupName);
        this.producer.setNamesrvAddr(namesrvAddr);
        this.producer.setMaxMessageSize(maxMessageSize);
        this.producer.setSendMsgTimeout(sendMsgTimeout);
        try {
            this.producer.start();
            LOGGER.info(String.format("producer is start! groupName:{%s},namesrvAddr:{%s}", this.groupName,this.namesrvAddr));
        } catch (MQClientException e) {
            LOGGER.error(String.format("producer error! groupName:{%s},namesrvAddr:{%s}", this.groupName,this.namesrvAddr,e));
            throw new TradeMQException(e);
        }
    }

    /**
     * 发送消息
     * @param topic 主题
     * @param tags  标签
     * @param keys  key
     * @param messageText   消息内容
     * @return  发送结果
     * @throws TradeMQException
     */
    private SendResult sendMessage(String topic,String tags,String keys,String messageText) throws TradeMQException{
        if(StringUtils.isBlank(topic)){
            throw new TradeMQException("topic is blank!");
        }
        if(StringUtils.isBlank(messageText)){
            throw new TradeMQException("messageText is blank!");
        }
        Message message=new Message(topic,tags,keys,messageText.getBytes());
        try {
            SendResult sendResult=this.producer.send(message);
            return sendResult;
        } catch (Exception e) {
            LOGGER.error("send message error:",e.getMessage());
            throw new TradeMQException(e);
        }


    }
    public SendResult sendMessage(MQEnums.TopicEnum topicEnum , String keys, String messageText) throws TradeMQException{
        return this.sendMessage(topicEnum.getTopic(),topicEnum.getTag(), keys, messageText);
    }
}
